package com.klm.easymq.aspect;

import com.klm.easymq.annotation.MQListener;
import com.klm.easymq.annotation.Retry;
import com.klm.easymq.core.IdempotentHandler;
import com.klm.easymq.core.IdempotentKeyGenerator;
import com.klm.easymq.core.MonitorService;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.DefaultParameterNameDiscoverer;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.expression.ExpressionParser;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.stereotype.Component;

import java.lang.reflect.Method;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * MQ监听器切面
 * 处理@MQListener注解的消息消费逻辑
 */
@Aspect
@Component
@Slf4j
public class MQListenerAspect {

    @Autowired
    private IdempotentHandler idempotentHandler;

    @Autowired
    private IdempotentKeyGenerator idempotentKeyGenerator;

    @Autowired
    private MonitorService monitorService;

    private final ExpressionParser expressionParser = new SpelExpressionParser();
    private final DefaultParameterNameDiscoverer parameterNameDiscoverer = new DefaultParameterNameDiscoverer();

    @Around("@annotation(mqListener)")
    public Object aroundMQListener(ProceedingJoinPoint joinPoint, MQListener mqListener) throws Throwable {
        Method method = ((MethodSignature) joinPoint.getSignature()).getMethod();
        Object[] args = joinPoint.getArgs();
        
        // 生成traceId
        String traceId = generateTraceId();
        
        // 记录消费开始
        monitorService.recordMessageConsume(traceId, mqListener.topic(), "CONSUMING", null);
        
        try {
            // 幂等控制
            if (!mqListener.idempotentKey().isEmpty()) {
                String idempotentKey = generateIdempotentKey(mqListener.idempotentKey(), method, args);
                if (!idempotentHandler.tryAcquire(idempotentKey)) {
                    log.warn("消息已处理，跳过重复消费: topic={}, traceId={}, key={}", 
                            mqListener.topic(), traceId, idempotentKey);
                    monitorService.recordMessageConsume(traceId, mqListener.topic(), "IDEMPOTENT_SKIP", null);
                    return null;
                }
            }
            
            // 执行重试逻辑
            Retry retryPolicy = mqListener.retryPolicy();
            return executeWithRetry(joinPoint, retryPolicy, traceId, mqListener.topic());
            
        } catch (Exception e) {
            log.error("消息消费失败: topic={}, traceId={}", mqListener.topic(), traceId, e);
            monitorService.recordMessageConsume(traceId, mqListener.topic(), "FAILED", e.getMessage());
            throw e;
        }
    }

    /**
     * 执行重试逻辑
     */
    private Object executeWithRetry(ProceedingJoinPoint joinPoint, Retry retryPolicy, 
                                   String traceId, String topic) throws Throwable {
        int maxAttempts = retryPolicy.maxAttempts();
        long interval = retryPolicy.interval();
        long maxInterval = retryPolicy.maxInterval();
        double multiplier = retryPolicy.multiplier();
        
        AtomicInteger attemptCount = new AtomicInteger(0);
        long currentInterval = interval;
        
        while (true) {
            try {
                attemptCount.incrementAndGet();
                log.debug("执行消息消费: topic={}, traceId={}, attempt={}", 
                         topic, traceId, attemptCount.get());
                
                Object result = joinPoint.proceed();
                
                // 消费成功
                monitorService.recordMessageConsume(traceId, topic, "SUCCESS", null);
                log.info("消息消费成功: topic={}, traceId={}, attempt={}", 
                        topic, traceId, attemptCount.get());
                
                return result;
                
            } catch (Exception e) {
                // 检查是否需要重试
                if (!shouldRetry(e, retryPolicy) || attemptCount.get() >= maxAttempts) {
                    log.error("消息消费最终失败: topic={}, traceId={}, attempt={}", 
                             topic, traceId, attemptCount.get(), e);
                    monitorService.recordMessageConsume(traceId, topic, "FINAL_FAILED", e.getMessage());
                    throw e;
                }
                
                // 计算下次重试间隔
                long nextInterval = calculateNextInterval(currentInterval, retryPolicy.backoffPolicy(), 
                                                        multiplier, maxInterval);
                
                log.warn("消息消费失败，准备重试: topic={}, traceId={}, attempt={}, nextInterval={}ms", 
                        topic, traceId, attemptCount.get(), nextInterval, e);
                
                // 等待重试
                Thread.sleep(nextInterval);
                currentInterval = nextInterval;
            }
        }
    }

    /**
     * 判断是否需要重试
     */
    private boolean shouldRetry(Exception e, Retry retryPolicy) {
        Class<? extends Throwable>[] retryableExceptions = retryPolicy.retryableExceptions();
        Class<? extends Throwable>[] noRetryExceptions = retryPolicy.noRetryExceptions();
        
        // 检查不重试异常
        for (Class<? extends Throwable> noRetryException : noRetryExceptions) {
            if (noRetryException.isAssignableFrom(e.getClass())) {
                return false;
            }
        }
        
        // 检查可重试异常
        if (retryableExceptions.length > 0) {
            for (Class<? extends Throwable> retryableException : retryableExceptions) {
                if (retryableException.isAssignableFrom(e.getClass())) {
                    return true;
                }
            }
            return false;
        }
        
        // 默认重试所有异常
        return true;
    }

    /**
     * 计算下次重试间隔
     */
    private long calculateNextInterval(long currentInterval, Retry.BackoffPolicy backoffPolicy, 
                                     double multiplier, long maxInterval) {
        switch (backoffPolicy) {
            case FIXED:
                return currentInterval;
            case EXPONENTIAL:
                long nextInterval = (long) (currentInterval * multiplier);
                return Math.min(nextInterval, maxInterval);
            case RANDOM:
                return (long) (currentInterval * (0.5 + Math.random()));
            default:
                return currentInterval;
        }
    }

    /**
     * 生成幂等键
     */
    private String generateIdempotentKey(String expression, Method method, Object[] args) {
        try {
            String[] parameterNames = parameterNameDiscoverer.getParameterNames(method);
            EvaluationContext context = new StandardEvaluationContext();
            
            if (parameterNames != null) {
                for (int i = 0; i < parameterNames.length && i < args.length; i++) {
                    context.setVariable(parameterNames[i], args[i]);
                }
            }
            
            Expression exp = expressionParser.parseExpression(expression);
            Object result = exp.getValue(context);
            return result != null ? result.toString() : UUID.randomUUID().toString();
            
        } catch (Exception e) {
            log.warn("生成幂等键失败，使用随机UUID: expression={}", expression, e);
            return UUID.randomUUID().toString();
        }
    }

    /**
     * 生成追踪ID
     */
    private String generateTraceId() {
        return UUID.randomUUID().toString();
    }
} 